/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

/*
 * MapperFor3rdLstHomoThm2ndPh.java
 *
 * Created on Jul 13, 2010, 4:09:45 PM
 */
package homomorphism.NewAPI;

import homomorphism.HomomorphismBase;
import utility.IntPair;
import utility.LongPair;
import java.io.IOException;
import java.lang.reflect.*;
import java.util.ArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapreduce.Mapper;

/**
 *
 * @author takeshi
 */

/*
 *  IdentityMapper
 *
 */
public class MapperFor3rdLstHomoThm2ndPh<InVal, OtVal> extends Mapper<LongPair, InVal, LongPair, OtVal> {

    private static final Log LOG = LogFactory.getLog(MapperFor3rdLstHomoThm2ndPh.class);
    private static final String userAppProperty = HomomorphismBase.UserClassNameProperty;
    // private Class FoldingType;
    private Class UnFoldingType;
    private String UserClassName;

    @Override
    public void setup(Context context) throws IOException, InterruptedException {

        UserClassName = context.getConfiguration().get(userAppProperty);
        //   FoldingType = job.getMapOutputValueClass();
        UnFoldingType = context.getOutputValueClass();
    }

    public ArrayList<OtVal> unfolding(InVal value) throws Exception {
        ClassLoader myloader = MapperFor3rdLstHomoThm2ndPh.class.getClassLoader();
        Class UserApp;
        if (UserClassName == null || UserClassName.isEmpty()) {
            return null;
        }
        try {
            UserApp = myloader.loadClass(UserClassName);
            Class partypes[] = new Class[1];
            partypes[0] = value.getClass(); // UnFoldingType;
            //for debug/test
            assert partypes[0].getSimpleName().equalsIgnoreCase(UnFoldingType.getSimpleName()) : "Something wrong, type not matched! ";
            Method _unfold = UserApp.getMethod("unfold", partypes);
            Object arglist[] = new Object[1];
            arglist[0] = value;
            Constructor ct = UserApp.getConstructor();//partypes
            Object methobj = ct.newInstance();
            ArrayList<OtVal> rst = (ArrayList<OtVal>) _unfold.invoke(methobj, arglist);
            return rst;
        } catch (ClassNotFoundException ee) {
            LOG.error(ee.getMessage());
        }
        return null;
    }

    @Override
    protected void map(LongPair key, InVal value, Context context)
            throws IOException, InterruptedException {

        // key(pid,index)
        try {
            ArrayList<OtVal> local = unfolding(value);

            int cout = 0;
            for (OtVal e : local) {
                LongPair newKey = new LongPair(0l, key.getSecond() + cout);
                context.write(newKey, e);
                LOG.info("2nd map : key is " + newKey + " val is " + e);
                cout++;
            }

        } catch (Exception ex) {
            LOG.error(ex.toString());
        }
    }
}
